package com.atguigu.realtime.streaming.apps

import java.sql.{Connection, PreparedStatement, ResultSet}
import java.time.{LocalDate, LocalDateTime}
import java.time.format.DateTimeFormatter
import java.util
import java.util.Properties

import com.alibaba.fastjson.JSON
import com.atguigu.realtime.constants.{PrefixConstant, TopicConstant}
import com.atguigu.realtime.streaming.apps.AlertApp.{appName, batchDuration, context, groupId, runApp, topic}
import com.atguigu.realtime.streaming.beans.{OrderDetail, OrderInfo, ProvinceInfo, SaleDetail, UserInfo}
import com.atguigu.realtime.streaming.utils.{JDBCUtil, MyKafkaUtil, PropertiesUtil, RedisUtil}
import com.google.gson.Gson
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.{CanCommitOffsets, HasOffsetRanges, OffsetRange}
import redis.clients.jedis.Jedis

import scala.collection.mutable
import scala.collection.mutable.ListBuffer

/**
 * Created by Smexy on 2022/5/28
 *
 * /*
 * 数据特征:
 * 事实表:
 * order_detail	: N
 * order_info	: 1
 * 有关联关系。都是用户下单后就产生。伴生关系。
 *
 * 当前时刻新增的订单，可能是一个历史用户(不是今天的新用户)下单的。
 * 在关联用户时，需要频繁地向Mysql的user_info表发送读请求，为了减轻Mysql服务器的压力，
 * 提升查询性能，可以使用缓存，将mysql中的user_info信息缓存在Redis。
 *
 *
 * 维度表:
 * user_info: 会经常新增及变化(修改)
 * 相对多
 *
 * 以上三张表，需要在程序中实时读取。需要先将以上三张表的数据采集到Kafka!
 *
 *
 * base_province: 基本是一个不变(不会发生写操作，只读)的维度表。
 * 数据量小。
 * 只读变量在Spark可以使用广播变量存储。
 * 如果不使用广播变量，一旦产生闭包，需要在Driver端将变量序列化给每一个Task。
 * 如果使用了广播变量，在Driver端将变量序列化给Executor.
 * Executor的数量大大小于Task,节省性能。
 * 要求： 变量必须只读。
 *
 *
 *
 **/
 *
 * `order_detail`： 购物详情，没有用户
 * LEFT JOIN
 * `order_info`： 用户id,省份id
 * LEFT JOIN
 * user_info :   用户信息
 * LEFT JOIN
 * `base_province`: 省份信息
 *
 * ----------------
 *  只有DS[K,V]才能Join，join时基于K进行关联。
 *
 *  ------------
 *   双流Join，数据无法Join的原因在于 要Join的数据无法在同一个批次被消费到。
 *
 *  ---------------------------
 *    解决：
 *          约定： 早到或晚到的数据都写入缓存，以供后续批次到达的数据进行关联。
 *
 *          对于order_info:
 *                          ①和当前批次已经到达的Order_detail关联
 *                          ②可能有早到的order_detail,去缓存中查询，查到就关联
 *                          ③无条件写入缓存，以防后续后晚到的order_detail
 *
 *          对于order_detail:
 *                          ①和当前批次已经到达的Order_Info关联
 *                          ②如果①无法关联，说明Order_detail早到或晚到。
 *                              进一步确认，取缓存找对应的Order_info，找到，就关联。
 *                              找不到，说明order_detail来早了，自己写入缓存，等待后续到达的order_info关联
 *
 *
 *   -----------------------------
 *      将order_info写入缓存，供后续的order-detail关联，根据orderId去查询对应的order_info.
 *          key:    order_info:orderid
 *          value:  String
 *
 *      将order_Detail存入缓存，供后续的Order_info关联，根据orderId查询对应的N个 order_detail
 *          key:    order_detail:orderid
 *           value:  Set()
 *
 *
 */
object SaleDetailApp extends BaseApp {
  override var appName: String = "SaleDetailApp"
  override var batchDuration: Int = 30
  override var groupId: String = "saledetail_order"
  override var topic: String = TopicConstant.ORDER_INFO
  val orderDetailTopic: String = TopicConstant.ORDER_DETAIL

  def main(args: Array[String]): Unit = {

    val properties: Properties = PropertiesUtil.load("config.properties")

    //设置ES的相关参数
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName(appName)

    sparkConf.set("es.nodes",properties.getProperty("es.nodes"))
    sparkConf.set("es.port",properties.getProperty("es.port"))
    sparkConf.set("es.index.auto.create", "true")

    context = new StreamingContext(sparkConf,Seconds(batchDuration))

    val provinceMap: Map[String, ProvinceInfo] = queryProvinceInfo()
    //广播省份信息
    val provinceBc: Broadcast[Map[String, ProvinceInfo]] = context.sparkContext.broadcast(provinceMap)

    runApp {

      //获取两个流
      val orderInfoDs: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(Array(topic), context, groupId)
      val orderDetailDs: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(Array(orderDetailTopic), context, groupId)

      //两个偏移量
      var orderInfoRanges: Array[OffsetRange] = null
      var orderDeetailRanges: Array[OffsetRange] = null

      //获取偏移量，封装Bean
      val ds1: DStream[(String, OrderInfo)] = orderInfoDs.transform(rdd => {

        orderInfoRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

        parseOrderInfo(rdd)

      })

      val ds2: DStream[(String, OrderDetail)] = orderDetailDs.transform(rdd => {

        orderDeetailRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

        rdd.map(record => {
          val orderDetail: OrderDetail = JSON.parseObject(record.value(), classOf[OrderDetail])
          (orderDetail.order_id, orderDetail)
        })

      })

      // 将流关联后，根据关联后的数据关联情况判断是否Join上了，再进行处理。保证关联后双方的数据都要保留
      val ds3: DStream[(String, (Option[OrderInfo], Option[OrderDetail]))] = ds1.fullOuterJoin(ds2)

      // 将 Order_info和order_detail关联后的数据
      val ds4: DStream[SaleDetail] = ds3.mapPartitions(partition => {

        val jedis: Jedis = RedisUtil.getJedisClient()

        //准备存放关联成功的SaleDetail的集合
        val result: ListBuffer[SaleDetail] = ListBuffer[SaleDetail]()

        val gson = new Gson()

        partition.foreach {
          case (orderId, (orderInfoOption, orderDetailOption)) => {

            if (orderInfoOption != None) {

              val orderInfo: OrderInfo = orderInfoOption.get

              if (orderDetailOption != None) {

                val orderDetail: OrderDetail = orderDetailOption.get

                // ①和当前批次已经到达的Order_detail关联
                result.append(new SaleDetail(orderInfo, orderDetail))

              }

              // ②可能有早到的order_detail,去缓存中查询，查到就关联
              //  如果是set集合，key不存在，返回Set()
              val orderDetailStrs: util.Set[String] = jedis.smembers(PrefixConstant.order_detail_redis_preffix + orderId)

              orderDetailStrs.forEach(str => {

                val detail: OrderDetail = JSON.parseObject(str, classOf[OrderDetail])

                result.append(new SaleDetail(orderInfo, detail))

              })

              // ③无条件写入缓存，以防后续后晚到的order_detail
              jedis.setex(PrefixConstant.order_info_redis_preffix + orderId, PrefixConstant.max_delay_time,gson.toJson(orderInfo))

            } else {

              // 在else中order_info为none，那么orderDetail一定不为none
              val orderDetail: OrderDetail = orderDetailOption.get

              //进一步确认，取缓存找对应的Order_info，找到，就关联。
              // 如果是string类型，key不存在，value是null
              val orderInfoStr: String = jedis.get(PrefixConstant.order_info_redis_preffix + orderId)

              if (orderInfoStr != null) {

                result.append(new SaleDetail(JSON.parseObject(orderInfoStr, classOf[OrderInfo]), orderDetail))
              } else {

                //找不到，说明order_detail来早了，自己写入缓存，等待后续到达的order_info关联
                jedis.sadd(PrefixConstant.order_detail_redis_preffix + orderId, gson.toJson(orderDetail))

                jedis.expire(PrefixConstant.order_detail_redis_preffix + orderId,PrefixConstant.max_delay_time)

              }
            }
          }
        }

        jedis.close()

        result.iterator


      })

      // 关联用户和省份
      val result: DStream[SaleDetail] = ds4.mapPartitions(partition => {

        val jedis: Jedis = RedisUtil.getJedisClient()

        val saleDetails: Iterator[SaleDetail] = partition.map(saleDetail => {

          val userStr: String = jedis.get(PrefixConstant.user_info_redis_preffix + saleDetail.user_id)

          val userInfo: UserInfo = JSON.parseObject(userStr, classOf[UserInfo])

          //关联用户信息
          saleDetail.mergeUserInfo(userInfo)

          //关联省份信息
          saleDetail.mergeProvinceInfo(provinceBc.value.get(saleDetail.province_id).get)

          saleDetail

        })

        jedis.close()

        saleDetails

      })

      import org.elasticsearch.spark._

      result.foreachRDD(rdd => {

        println("即将写入:"+rdd.count())

        rdd.saveToEs("realtime2022_sale_detail_"+LocalDate.now+"/_doc",Map("es.mapping.id" -> "order_detail_id"))

        //提交偏移量
        orderInfoDs.asInstanceOf[CanCommitOffsets].commitAsync(orderInfoRanges)
        orderDetailDs.asInstanceOf[CanCommitOffsets].commitAsync(orderDeetailRanges)

      })

    }


  }

  //查询省份信息
  def queryProvinceInfo():Map[String,ProvinceInfo]={

    val result = new mutable.HashMap[String, ProvinceInfo]()

    val sql=
      """
        |
        |select
        |   *
        |from base_province
        |
        |
        |""".stripMargin

    var conn:Connection = null
    var ps: PreparedStatement = null

    try {
      conn = JDBCUtil.getConnection()
      ps = conn.prepareStatement(sql)

      val resultSet: ResultSet = ps.executeQuery()
      while (resultSet.next()) {

        //当前组上次提交的消费当前主题的每一个分区的偏移量
        result.put( resultSet.getLong("id").toString, ProvinceInfo(
          resultSet.getLong("id").toString,
          resultSet.getString("name"),
          resultSet.getString("region_id"),
          resultSet.getString("area_code"),
          resultSet.getString("iso_code"),
          resultSet.getString("iso_3166_2")
        ))

      }
    } catch {
      case e:Exception =>e.printStackTrace()
    } finally {

      if(ps != null){
        ps.close()

      }

      if(conn != null){

        conn.close()
      }

    }
    result.toMap

  }

  def parseOrderInfo(rdd: RDD[ConsumerRecord[String, String]]):RDD[(String,OrderInfo)] = {

    rdd.mapPartitions(partition => {

      //声明时间格式
      val formatter1: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd")
      val formatter3: DateTimeFormatter = DateTimeFormatter.ofPattern("HH")
      val formatter2: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")

      partition.map(record => {

        val orderInfo: OrderInfo = JSON.parseObject(record.value(), classOf[OrderInfo])

        // create_time": "2022-05-25 02:23:00
        //为create_date和create_hour赋值
        val time: LocalDateTime = LocalDateTime.parse(orderInfo.create_time, formatter2)
        orderInfo.create_date = time.format(formatter1)
        orderInfo.create_hour = time.format(formatter3)

        (orderInfo.id,orderInfo)

      })

    })

  }
}
